package com.kenliang.netty.rpc.netty;

import com.google.common.util.concurrent.*;
import com.kenliang.netty.rpc.config.RpcSystemConfig;
import com.kenliang.netty.rpc.parallel.RpcThreadPool;
import com.kenliang.netty.rpc.spring.SpringUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.checkerframework.checker.nullness.qual.Nullable;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author KenLiang
 * @description:客户端加载类
 * @date Created in 4:59 PM 2019/3/21.
 */
@Slf4j
public class RpcClientLoader {

    private MessageSendHandler messageSendHandler;

    private ReentrantLock lock = new ReentrantLock();
    private Condition connectStatus = lock.newCondition();

    private static ListeningExecutorService threadPoolExecutor = MoreExecutors.listeningDecorator(
            (ThreadPoolExecutor) RpcThreadPool.getExecutor(RpcSystemConfig.CLIENT_THREAD_POOL_THREAD_NUM, RpcSystemConfig.CLIENT_THREAD_POOL_QUEUE_NUM));

    public static class RpcClientLoaderHolder {
        private static RpcClientLoader INSTANCE = new RpcClientLoader();
    }


    private RpcClientLoader() {

    }

    public static RpcClientLoader getInstance() {
        return RpcClientLoaderHolder.INSTANCE;

    }

    public MessageSendHandler getMessageSendHandler() {
        try {
            lock.lock();
            if (this.messageSendHandler == null) {
                connectStatus.await();
            }
        } catch (InterruptedException e) {
            log.error(ExceptionUtils.getStackTrace(e));
        } finally {
            lock.unlock();
        }
        return this.messageSendHandler;
    }

    public void setMessageSendHandler(MessageSendHandler messageSendHandler) {
        try {
            lock.lock();
            this.messageSendHandler = messageSendHandler;
            connectStatus.signalAll();
        }finally {
            lock.unlock();
        }

    }


    public void load(String serverAddress) {
        String[] address = serverAddress.split(":");
        //todo 兼容配置多个服务？？
        if (address.length >= 2) {
            try {
                InetSocketAddress inetSocketAddress = new InetSocketAddress(address[0], Integer.valueOf(address[1]));
                ListenableFuture<Boolean> future = threadPoolExecutor.submit(new MessageSendInitializeTask(inetSocketAddress));
//                Futures.addCallback(future, new FutureCallback<Boolean>() {
//                    @Override
//                    public void onSuccess(@Nullable Boolean result) {
//                        log.info("FutureCallback onSuccess");
//                    }
//
//                    @Override
//                    public void onFailure(Throwable t) {
//                        log.error("FutureCallback onFailure",ExceptionUtils.getStackTrace(t));
//                    }
//                }, threadPoolExecutor);
            } catch (Exception e) {
                log.error("调用者连接失败：", ExceptionUtils.getStackTrace(e));
            }
        } else {
            log.error("server地址错误：{}", address);
        }

    }
}
